Smart conn #637
Merged
Smart conn #637
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
added 7 commits
June 2, 2017 16:41
Conflicts: store
Conflicts: store
Conflicts: store
sumwale
reviewed
Jun 3, 2017
Contributor
sumwale
left a comment
sumwale
left a comment
There was a problem hiding this comment.
See some comments. About point #3 have a doubt as to how that will work properly if we are not ensuring that commit happens after all tasks are done. Perhaps a separate message at the end or some accounting on executor side keeping track of partitions scheduled on that executor so that it will commit only if all the partitions are done.
| SparkShellRDDHelper.snapshotTxId.set(txid) | ||
| val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID (?)") | ||
| getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR) | ||
| getSnapshotTXId.execute |
Contributor
There was a problem hiding this comment.
use empty parens in actions as per scala convention (and none in getters)
| val txid: String = getSnapshotTXId.getString(1) | ||
| getSnapshotTXId.close() | ||
| SparkShellRDDHelper.snapshotTxId.set(txid) | ||
| logDebug(s"The snapshot tx id is ${txid} and tablename is ${tableName}") |
Contributor
There was a problem hiding this comment.
indentation change looks incorrect and older one with 2 spaces was proper
| ps.close() | ||
| SparkShellRDDHelper.snapshotTxId.set(null) | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
Shouldn't this be done after all tasks are done? Won't a commit in the middle of another task execution cause trouble?
Contributor
Author
There was a problem hiding this comment.
Yes in general. But for read operations if the txState is already being used in an iterator , even if the state is closed, we compare with the snapshot stored.
For write operations this would cause issue if tx are committed by a task prematurely, however currently I haven't found a scenario for that in this task.
Contributor
There was a problem hiding this comment.
@suranjan
Even then a read in a collocated join may not even be scheduled (or scheduled by Spark but not by OS) while another task commits. In that case the two will use different snapshots? I don't see what kind of semantics this change is trying to provide.
Contributor
|
- What are the semantics of concurrent tasks/threads participating in the
same txn? Couldn't this result in possible loss of data integrity? Unless,
the entire state across all tasks is accounted for as a unit and a single
thread executes the commit protocol.
- Are there implications when spark is re-executing tasks? i.e. have we
accounted for all cases when tasks could get re-executed by Spark.
…-----
Jags
SnappyData blog <http://www.snappydata.io/blog>
Download binary, source <https://github.com/SnappyDataInc/snappydata>
On Sat, Jun 3, 2017 at 4:03 AM, Sumedh Wale ***@***.***> wrote:
***@***.**** commented on this pull request.
See some comments. About point #3
<#3> have a doubt as to
how that will work properly if we are not ensuring that commit happens
after all tasks are done. Perhaps a separate message at the end or some
accounting on executor side keeping track of partitions scheduled on that
executor so that it will commit only if all the partitions are done.
------------------------------
In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
JDBCSourceAsColumnarStore.scala
<#637 (comment)>
:
> val rs = stmt.executeQuery()
// get the txid which was used to take the snapshot.
if (!_commitTx) {
- val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID (?)")
- getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
- getSnapshotTXId.execute
- val txid: String = getSnapshotTXId.getString(1)
- getSnapshotTXId.close()
- SparkShellRDDHelper.snapshotTxId.set(txid)
+ val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID (?)")
+ getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
+ getSnapshotTXId.execute
use empty parens in actions as per scala convention (and none in getters)
------------------------------
In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
JDBCSourceAsColumnarStore.scala
<#637 (comment)>
:
> val rs = stmt.executeQuery()
// get the txid which was used to take the snapshot.
if (!_commitTx) {
- val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID (?)")
- getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
- getSnapshotTXId.execute
- val txid: String = getSnapshotTXId.getString(1)
- getSnapshotTXId.close()
- SparkShellRDDHelper.snapshotTxId.set(txid)
+ val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID (?)")
+ getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
+ getSnapshotTXId.execute
+ val txid: String = getSnapshotTXId.getString(1)
+ getSnapshotTXId.close()
+ SparkShellRDDHelper.snapshotTxId.set(txid)
+ logDebug(s"The snapshot tx id is ${txid} and tablename is ${tableName}")
indentation change looks incorrect and older one with 2 spaces was proper
------------------------------
In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
JDBCSourceAsColumnarStore.scala
<#637 (comment)>
:
> @@ -595,6 +596,24 @@ class SmartConnectorRowRDD(_session: SnappySession,
pushProjections = true, useResultSet = true, _connProperties,
_filters, _partEval, _commitTx) {
+
+ override def commitTxBeforeTaskCompletion(conn: Option[Connection], context: TaskContext) = {
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => {
+ val txId = SparkShellRDDHelper.snapshotTxId.get
+ logDebug(s"The txid going to be committed is $txId " + tableName)
+ if ((txId ne null) && !txId.equals("null")
+ /* && !(tx.asInstanceOf[TXStateProxy]).isClosed() */ ) {
+ val ps = conn.get.prepareStatement(s"call sys.COMMIT_SNAPSHOT_TXID(?)")
+ ps.setString(1, txId)
+ ps.executeUpdate()
+ logDebug(s"The txid being committed is $txId")
+ ps.close()
+ SparkShellRDDHelper.snapshotTxId.set(null)
+ }
+ }
Shouldn't this be done *after* all tasks are done? Won't a commit in the
middle of another task execution cause trouble?
—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
<#637 (review)>,
or mute the thread
<https://github.com/notifications/unsubscribe-auth/AB2KBqXi_a0OE2F1t0abIHstJu4qDZnpks5sAT18gaJpZM4Nu94y>
.
|
Contributor
Author
|
Currently, the snapshot is not across task. Each task takes its own
snapshot, so snapshot isolation is at partition level. Once we have
lead/driver initiated transaction with global snapshot implementation, we
can provide snapshot across tasks.
Because we take snapshot when a task starts, if a task is re-executed then
snapshot will also be new.
On Sat, Jun 3, 2017 at 8:05 PM, Jags Ramnarayan <[email protected]>
wrote:
… - What are the semantics of concurrent tasks/threads participating in the
same txn? Couldn't this result in possible loss of data integrity? Unless,
the entire state across all tasks is accounted for as a unit and a single
thread executes the commit protocol.
- Are there implications when spark is re-executing tasks? i.e. have we
accounted for all cases when tasks could get re-executed by Spark.
-----
Jags
SnappyData blog <http://www.snappydata.io/blog>
Download binary, source <https://github.com/SnappyDataInc/snappydata>
On Sat, Jun 3, 2017 at 4:03 AM, Sumedh Wale ***@***.***>
wrote:
> ***@***.**** commented on this pull request.
>
> See some comments. About point #3
> <#3> have a doubt as to
> how that will work properly if we are not ensuring that commit happens
> after all tasks are done. Perhaps a separate message at the end or some
> accounting on executor side keeping track of partitions scheduled on that
> executor so that it will commit only if all the partitions are done.
> ------------------------------
>
> In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
> JDBCSourceAsColumnarStore.scala
> <https://github.com/SnappyDataInc/snappydata/pull/
637#discussion_r119984539>
> :
>
> > val rs = stmt.executeQuery()
>
> // get the txid which was used to take the snapshot.
> if (!_commitTx) {
> - val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID
(?)")
> - getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
> - getSnapshotTXId.execute
> - val txid: String = getSnapshotTXId.getString(1)
> - getSnapshotTXId.close()
> - SparkShellRDDHelper.snapshotTxId.set(txid)
> + val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID
(?)")
> + getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
> + getSnapshotTXId.execute
>
> use empty parens in actions as per scala convention (and none in getters)
> ------------------------------
>
> In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
> JDBCSourceAsColumnarStore.scala
> <https://github.com/SnappyDataInc/snappydata/pull/
637#discussion_r119984554>
> :
>
> > val rs = stmt.executeQuery()
>
> // get the txid which was used to take the snapshot.
> if (!_commitTx) {
> - val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID
(?)")
> - getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
> - getSnapshotTXId.execute
> - val txid: String = getSnapshotTXId.getString(1)
> - getSnapshotTXId.close()
> - SparkShellRDDHelper.snapshotTxId.set(txid)
> + val getSnapshotTXId = conn.prepareCall(s"call sys.GET_SNAPSHOT_TXID
(?)")
> + getSnapshotTXId.registerOutParameter(1, java.sql.Types.VARCHAR)
> + getSnapshotTXId.execute
> + val txid: String = getSnapshotTXId.getString(1)
> + getSnapshotTXId.close()
> + SparkShellRDDHelper.snapshotTxId.set(txid)
> + logDebug(s"The snapshot tx id is ${txid} and tablename is
${tableName}")
>
> indentation change looks incorrect and older one with 2 spaces was proper
> ------------------------------
>
> In core/src/main/scala/org/apache/spark/sql/execution/columnar/impl/
> JDBCSourceAsColumnarStore.scala
> <https://github.com/SnappyDataInc/snappydata/pull/
637#discussion_r119984578>
> :
>
> > @@ -595,6 +596,24 @@ class SmartConnectorRowRDD(_session:
SnappySession,
> pushProjections = true, useResultSet = true, _connProperties,
> _filters, _partEval, _commitTx) {
>
> +
> + override def commitTxBeforeTaskCompletion(conn: Option[Connection],
context: TaskContext) = {
> + Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => {
> + val txId = SparkShellRDDHelper.snapshotTxId.get
> + logDebug(s"The txid going to be committed is $txId " + tableName)
> + if ((txId ne null) && !txId.equals("null")
> + /* && !(tx.asInstanceOf[TXStateProxy]).isClosed() */ ) {
> + val ps = conn.get.prepareStatement(s"call
sys.COMMIT_SNAPSHOT_TXID(?)")
> + ps.setString(1, txId)
> + ps.executeUpdate()
> + logDebug(s"The txid being committed is $txId")
> + ps.close()
> + SparkShellRDDHelper.snapshotTxId.set(null)
> + }
> + }
>
> Shouldn't this be done *after* all tasks are done? Won't a commit in the
> middle of another task execution cause trouble?
>
> —
> You are receiving this because you are subscribed to this thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/SnappyDataInc/snappydata/pull/
637#pullrequestreview-41918181>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AB2KBqXi_
a0OE2F1t0abIHstJu4qDZnpks5sAT18gaJpZM4Nu94y>
> .
>
—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
<#637 (comment)>,
or mute the thread
<https://github.com/notifications/unsubscribe-auth/AALmGcgm-M-kn-iezZvFYumQ7ouOD0TPks5sAW9KgaJpZM4Nu94y>
.
--
Regards,
Suranjan Kumar
SnappyData (http://www.snappydata.io)
|
added 2 commits
June 3, 2017 22:33
Conflicts: store
Contributor
|
@suranjan If we are providing paritition-level isolation semantics only as you mention, then what is this PR trying to do using TX across tasks (which will be one per partition)? What is this trying to solve? This will result in inconsistent semantics. |
Contributor
Author
|
If the multiple tasks are being executed by the same thread, it will use
the same snapshot even if the first task has completed. for e.g (row buffer
scan + column Table scan), similarly for colocated join.
May be my understaning of colocated join is wrong, but in the log I saw
both tables being iterated simultaneously in the same thread on spark side.
For, this case, in embedded mode we do use same snapshot. For smart
connector mode, different snapshots were being used due to different
connection object and one was being overwritten by other in the threadlocal
variable on spark side.
…On Sun, Jun 4, 2017 at 12:17 AM, Sumedh Wale ***@***.***> wrote:
@suranjan <https://github.com/suranjan> If we are providing
paritition-level isolation semantics only as you mention, then what is this
PR trying to do using TX across tasks (which will be one per partition)?
What is this trying to solve? This will result in inconsistent semantics.
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
<#637 (comment)>,
or mute the thread
<https://github.com/notifications/unsubscribe-auth/AALmGYIVA5MDfmP4dZQf5qZhNvxwHthRks5sAao-gaJpZM4Nu94y>
.
--
Regards,
Suranjan Kumar
SnappyData (http://www.snappydata.io)
|
Contributor
|
Row buffer scan + column buffer scan of a single bucket is a single task. Likewise collocated join of two buckets is a single task. A single thread executing multiple tasks in same job should take separate snapshots for reads to keep clean semantics. Even for writes, row buffer writes and column buffer writes are all same task though different code segments. Anyway since this is already merged lets discuss what to do going forward later (unless this change causes trouble elsewhere). |
Contributor
Author
|
Ok.
The pull request covers both
1. Colocate join of two buckets (in case of smart connector)
2. Single thread executing multiple tasks in same job.
Without this merge, 1st was broken as multiple scan in smart connector mode
was using multiple snapshot.
…On Sun, Jun 4, 2017 at 12:46 AM, Sumedh Wale ***@***.***> wrote:
Row buffer scan + column buffer scan of a single bucket is a single task.
Likewise collocated join of two buckets is a single task. A single thread
executing multiple tasks in same job should take separate snapshots for
reads to keep clean semantics. Even for writes, row buffer writes and
column buffer writes are all same task though different code segments.
Anyway since this is already merged lets discuss what to do going forward
later (unless this change causes trouble elsewhere).
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
<#637 (comment)>,
or mute the thread
<https://github.com/notifications/unsubscribe-auth/AALmGdn0JRAi60vviIgHSMI9t9xvvzE5ks5sAbDygaJpZM4Nu94y>
.
--
Regards,
Suranjan Kumar
SnappyData (http://www.snappydata.io)
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Changes proposed in this pull request
Patch testing
precheckin
ReleaseNotes.txt changes
(Does this change require an entry in ReleaseNotes.txt? If yes, has it been added to it?)
Other PRs
(Does this change require changes in other projects- store, spark, spark-jobserver, aqp? Add the links of PR of the other subprojects that are related to this change)